package com.icarbonx.dap.notification.job

import com.google.gson.Gson
import com.icarbonx.dap.notification.msgservice.rabbitmq.MQObject
import com.icarbonx.dap.notification.msgservice.redis.RedisEndpoint
import com.icarbonx.dap.notification.util.{KafkaUtil, SparkUtil, DateUtil, NoticeMsg}
import com.icarbonx.dap.notification.vo._
import kafka.serializer.StringDecoder
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.log4j.{Level, Logger}
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Duration, Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayBuffer

/**
 * Created by Lenovo on 2018/8/12.
 * 一天内饮食后1小时,血糖连续2个点低于3, 可以存储多天 作为多发 ,可以多发 访问多天
 * 用餐后\午餐\晚餐\
 * 多发5天内:餐后\午餐\晚餐\
 */
object BloodSugarFoodMealsStream {
  def main(args: Array[String]) {
    Logger.getLogger("org.apache.kafka").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.zookeeper").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

    /**
     * brokers = "10.47.250.4:9092,10.47.250.7:9092,10.47.250.6:9092"
     * topicList = "abel_test"
     * mq:10.104.103.7 5672 guest icx@betaOn16 /
     * redis:10.104.103.7 6379 passxx#$#dslllfjshwerw123@ 3000 abel_test
     */
    println(args.mkString("\t"))
    val Array(
    brokers,
    topicList,
    otherTopicList,
    mqHost,
    mqPort,
    mqUsername,
    mqPassword,
    mqVirtualHost,
    exchangeName,
    mqRouteKey,
    duation, appName
    ) = args
    val conf = SparkUtil.getSparkConf(appName, Array(classOf[BloodSugarRecord], classOf[FoodRecord], classOf[MealRecord]))
    //      new SparkConf().setAppName("BloodSugarDrugStream")
    //     //每秒钟每个分区kafka拉取消息的速率
    //      .set("spark.streaming.kafka.maxRatePerPartition", "100")
    val sc = SparkContext.getOrCreate(conf)
    val checkpointDir =
      "/data/streaming/checkpoint/blood_sugar_foodmeals_stream/"

    val updateSugarFunc = (values: Seq[ArrayBuffer[BloodSugarRecord]],
                           state: Option[ArrayBuffer[BloodSugarRecord]]) => {
      //保存状态
      val currentCount = ArrayBuffer[BloodSugarRecord]()
      values.foreach(list => currentCount ++= list)
      val previousCount = state.getOrElse(ArrayBuffer[BloodSugarRecord]())
      Some(currentCount ++ previousCount)
    }
    val updateDrugFunc = (values: Seq[ArrayBuffer[FoodRecord]],
                          state: Option[ArrayBuffer[FoodRecord]]) => {
      //保存状态
      val currentCount = ArrayBuffer[FoodRecord]()
      values.foreach(list => currentCount ++= list)
      val previousCount = state.getOrElse(ArrayBuffer[FoodRecord]())
      Some(currentCount ++ previousCount)
    }

    def creatingFunc(): StreamingContext = {
      val ssc = new StreamingContext(sc, Seconds(duation.toInt))
      ssc.checkpoint(checkpointDir)
      val kafkaParams = KafkaUtil.getKafkaParam(brokers, appName)

      //        Map(
      //        "metadata.broker.list" -> brokers,
      //        "group.id" -> "BloodSugarFoodMealsStream"
      //      )
      val mqConn = new MQObject(
        mqHost,
        mqPort.toInt,
        mqUsername,
        mqPassword,
        mqVirtualHost
      )


      val sugarTopics = topicList.split(",").toSet
      val foodTopics = otherTopicList.split(",") toSet
      val sugarMessages = KafkaUtils
        .createDirectStream[String, String, StringDecoder, StringDecoder](
          ssc,
          kafkaParams,
          sugarTopics
        )
      val foodMessages = KafkaUtils
        .createDirectStream[String, String, StringDecoder, StringDecoder](
          ssc,
          kafkaParams,
          foodTopics
        )

      val sugarBatch: DStream[(String, ArrayBuffer[BloodSugarRecord])] =
        sugarMessages
          .filter(_._2.nonEmpty)
          .flatMap { msg =>
            try {
              println("___________ msg :" + msg._2)
              val gson = new Gson()
              val voList =
                gson.fromJson(msg._2, classOf[Array[BloodSugarRecord]])
              voList.map { vo =>
                (vo.personId.toString, ArrayBuffer[BloodSugarRecord](vo))
              }.toList
            } catch {
              case e: Exception =>
                println("____________" + e.getMessage)
                List(("", ArrayBuffer[BloodSugarRecord]()))
            }
          }
          .filter(_._1.nonEmpty)
          .updateStateByKey(updateSugarFunc)

      val foodBatch: DStream[(String, ArrayBuffer[FoodRecord])] = foodMessages
        .filter(_._2.nonEmpty)
        .map { msg =>
          try {
            println("___________ msg :" + msg)
            val gson = new Gson()
            val vo = gson.fromJson(msg._2, classOf[FoodRecord])
            (vo.personId.toString, ArrayBuffer[FoodRecord](vo))
          } catch {
            case e: Exception =>
              println("____________" + e.getMessage)
              ("", ArrayBuffer[FoodRecord]())
          }
        }
        .filter(_._1.nonEmpty)
        .updateStateByKey(updateDrugFunc)

      val sortRddSugar = sugarBatch.map {
        case (pid, list) => (pid, list.sortBy(_.timestamp))
      }
      val sortRddFood = foodBatch.map {
        case (pid, list) => (pid, list.sortBy(_.ateTime))
      }

//      sortRddSugar.checkpoint(Duration(300 * 1000))
//      sortRddFood.checkpoint(Duration(300 * 1000))
      //val sugarAndFood = foodBatch.rightOuterJoin(sugarBatch)
      sortRddFood.foreachRDD(messD => {
        messD.foreachPartition(linesD => {
          linesD.foreach {
            case (pidD, listD) => {
              val theLastFoodTime = listD.last.ateTime
              sortRddSugar.foreachRDD(messS => {
                messS.foreachPartition(linesS => {
                  linesS.filter(a => a._1 == pidD).foreach {
                    case (pidS, listS) => {
                      val mqMsg = ArrayBuffer[String]()
                      val gson = new Gson()
                      println(
                        "____ pid :" + pidD + ",,, list:" + listD
                          .map(gson.toJson(_))
                          .toString()
                      )
                      val sugarTimeOneHour = theLastFoodTime - 60 * 60 * 1000
                      val sugarOneHour = listS.filter(
                        a => a.timestamp > theLastFoodTime && a.timestamp < sugarTimeOneHour
                      )
                      if (sugarOneHour.nonEmpty) {
                        var preVo = sugarOneHour.head
                        for (i <- 1 to sugarOneHour.size - 1) {
                          val nowVo = sugarOneHour(i)
                          //val preVo = lastHourList(i-1)
                          //纯血糖数据报警6个
                          if (nowVo.timestamp - preVo.timestamp < 15 * 60 * 1000) {
                            //低血糖
                            if (preVo.value > 12 && nowVo.value > 12) {
                              val msg = new WarningMsg(
                                "glycemia",
                                preVo.personId,
                                NoticeMsg.high_glycemia_meal._1,
                                NoticeMsg.high_glycemia_meal._2,
                                NoticeMsg.high_glycemia_meal._3,
                                DateUtil.getTodayDateTime().toString,
                                NoticeMsg.high_glycemia_meal._5,
                                NoticeMsg.high_glycemia_meal._4,
                                NoticeMsg.high_glycemia_meal._6
                              )
                              mqMsg += gson.toJson(msg)
                              println(
                                "______ gson.toJson(msg)" + gson.toJson(msg)
                              )
                            }
                            if (DateUtil.getDateTime(
                              DateUtil.getTodayDateTime()
                            )(3) == 12) {
                              if (preVo.value < 2.8 && nowVo.value < 2.8) {
                                val msg = new WarningMsg(
                                  "glycemia",
                                  sugarOneHour.last.personId,
                                  NoticeMsg.low_glycemia_lunch._1,
                                  NoticeMsg.low_glycemia_lunch._2,
                                  NoticeMsg.low_glycemia_lunch._3,
                                  DateUtil.getTodayDateTime().toString,
                                  NoticeMsg.low_glycemia_lunch._5,
                                  NoticeMsg.low_glycemia_lunch._4,
                                  NoticeMsg.low_glycemia_lunch._6
                                )
                                mqMsg += gson.toJson(msg)
                                println(
                                  "______ gson.toJson(msg)" + gson.toJson(msg)
                                )
                              }
                            }
                          }
                          preVo = nowVo
                        }
                        if (sugarOneHour
                          .maxBy(_.value)
                          .value - sugarOneHour.last.value > 6) {
                          val msg = new WarningMsg(
                            "glycemia",
                            sugarOneHour.last.personId,
                            NoticeMsg.slope_glycemia_meal._1,
                            NoticeMsg.slope_glycemia_meal._2,
                            NoticeMsg.slope_glycemia_meal._3,
                            DateUtil.getTodayDateTime().toString,
                            NoticeMsg.slope_glycemia_meal._5,
                            NoticeMsg.slope_glycemia_meal._4,
                            NoticeMsg.slope_glycemia_meal._6
                          )
                          mqMsg += gson.toJson(msg)
                          println(
                            "______ gson.toJson(msg)" + gson.toJson(msg)
                          )
                        }
                      }
                      val sugarTimeTwoHour = theLastFoodTime - 2 * 60 * 60 * 1000
                      val sugarTwoHour = listS.filter(
                        a => a.timestamp > sugarTimeTwoHour && a.timestamp < theLastFoodTime
                      )
                      if (sugarTwoHour.nonEmpty) {
                        var preVo = sugarTwoHour.head
                        for (i <- 1 to sugarTwoHour.size - 1) {
                          val nowVo = sugarOneHour(i)
                          if (DateUtil.getDateTime(
                            DateUtil.getTodayDateTime()
                          )(3) == 19) {
                            if (preVo.value < 2.8 && nowVo.value < 2.8) {
                              val msg = new WarningMsg(
                                "glycemia",
                                sugarOneHour.last.personId,
                                NoticeMsg.low_glycemia_supper._1,
                                NoticeMsg.low_glycemia_supper._2,
                                NoticeMsg.low_glycemia_supper._3,
                                DateUtil.getTodayDateTime().toString,
                                NoticeMsg.low_glycemia_supper._5,
                                NoticeMsg.low_glycemia_supper._4,
                                NoticeMsg.low_glycemia_supper._6
                              )
                              mqMsg += gson.toJson(msg)
                              println(
                                "______ gson.toJson(msg)" + gson.toJson(msg)
                              )
                            }
                          }
                        }
                      }
                      if (mqMsg.nonEmpty) {
                        mqConn.sendMsgs(mqMsg, exchangeName, mqRouteKey)
                      }
                      //                      if (mqMsg.nonEmpty) {
                      //                        val mqChannel = mqConn.connect().createChannel()
                      //                        mqMsg.foreach { msg =>
                      //                          // 声明exchange
                      //                          mqChannel.exchangeDeclare(exchangeName, "topic");
                      //                          // 发送消息, 指定RoutingKey
                      //                          mqChannel.basicPublish(
                      //                            exchangeName,
                      //                            mqRouteKey,
                      //                            null,
                      //                            msg.toString.getBytes("UTF-8")
                      //                          );
                      //                          println("____ msg :" + msg)
                      //                        }
                      //                        //            if (null != mqChannel) mqChannel.close()
                      //                      }
                    }
                  }
                })
              })
            }
          }
        })
      })
      ssc
    }


    val ssc = StreamingContext.getOrCreate(checkpointDir, creatingFunc)

    ssc.start()
    ssc.awaitTermination()
  }
}